{
 "cells": [
  {
   "cell_type": "code",
   "execution_count": null,
   "id": "28d8fbc7",
   "metadata": {},
   "outputs": [],
   "source": [
    "# | default_exp _server"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": null,
   "id": "7fcc07a5",
   "metadata": {},
   "outputs": [],
   "source": [
    "# | export\n",
    "\n",
    "import asyncio\n",
    "import multiprocessing\n",
    "import platform\n",
    "import signal\n",
    "import threading\n",
    "from contextlib import contextmanager\n",
    "from typing import *\n",
    "from types import FrameType\n",
    "\n",
    "import asyncer\n",
    "import typer\n",
    "\n",
    "from fastkafka._components.helpers import _import_from_string\n",
    "from fastkafka._components.logger import get_logger\n",
    "from fastkafka._components._subprocess import terminate_asyncio_process"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": null,
   "id": "170ddd1c",
   "metadata": {},
   "outputs": [],
   "source": [
    "import os\n",
    "from time import sleep\n",
    "\n",
    "from pydantic import BaseModel\n",
    "from typer.testing import CliRunner\n",
    "\n",
    "from fastkafka._components.logger import suppress_timestamps\n",
    "from fastkafka._components.test_dependencies import generate_app_in_tmp\n",
    "from fastkafka.testing import ApacheKafkaBroker"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": null,
   "id": "62aaa890",
   "metadata": {},
   "outputs": [],
   "source": [
    "# | notest\n",
    "\n",
    "# allows async calls in notebooks\n",
    "\n",
    "import nest_asyncio"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": null,
   "id": "55be7819",
   "metadata": {},
   "outputs": [],
   "source": [
    "# | notest\n",
    "\n",
    "nest_asyncio.apply()"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": null,
   "id": "de03595b",
   "metadata": {},
   "outputs": [],
   "source": [
    "# | export\n",
    "\n",
    "logger = get_logger(__name__, level=20)"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": null,
   "id": "b9ee8c91",
   "metadata": {},
   "outputs": [
    {
     "name": "stdout",
     "output_type": "stream",
     "text": [
      "[INFO] __main__: ok\n"
     ]
    }
   ],
   "source": [
    "suppress_timestamps()\n",
    "logger = get_logger(__name__, level=20)\n",
    "logger.info(\"ok\")"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": null,
   "id": "a2d2a997",
   "metadata": {},
   "outputs": [],
   "source": [
    "# | export\n",
    "\n",
    "\n",
    "class ServerProcess:\n",
    "    def __init__(self, app: str, kafka_broker_name: str):\n",
    "        \"\"\"\n",
    "        Represents a server process for running the FastKafka application.\n",
    "\n",
    "        Args:\n",
    "            app (str): Input in the form of 'path:app', where **path** is the path to a python file and **app** is an object of type **FastKafka**.\n",
    "            kafka_broker_name (str): The name of the Kafka broker, one of the keys of the kafka_brokers dictionary passed in the constructor of FastKafka class.\n",
    "        \"\"\"\n",
    "        self.app = app\n",
    "        self.should_exit = False\n",
    "        self.kafka_broker_name = kafka_broker_name\n",
    "\n",
    "    def run(self) -> None:\n",
    "        \"\"\"\n",
    "        Runs the FastKafka application server process.\n",
    "        \"\"\"\n",
    "        return asyncio.run(self._serve())\n",
    "\n",
    "    async def _serve(self) -> None:\n",
    "        \"\"\"\n",
    "        Internal method that runs the FastKafka application server.\n",
    "        \"\"\"\n",
    "        self._install_signal_handlers()\n",
    "\n",
    "        self.application = _import_from_string(self.app)\n",
    "        self.application.set_kafka_broker(self.kafka_broker_name)\n",
    "\n",
    "        async with self.application:\n",
    "            await self._main_loop()\n",
    "\n",
    "    def _install_signal_handlers(self) -> None:\n",
    "        \"\"\"\n",
    "        Installs signal handlers for handling termination signals.\n",
    "        \"\"\"\n",
    "        if threading.current_thread() is not threading.main_thread():\n",
    "            raise RuntimeError()\n",
    "\n",
    "        loop = asyncio.get_event_loop()\n",
    "\n",
    "        HANDLED_SIGNALS = (\n",
    "            signal.SIGINT,  # Unix signal 2. Sent by Ctrl+C.\n",
    "            signal.SIGTERM,  # Unix signal 15. Sent by `kill <pid>`.\n",
    "        )\n",
    "        if platform.system() == \"Windows\":\n",
    "            HANDLED_SIGNALS = (*HANDLED_SIGNALS, signal.SIGBREAK) # type: ignore\n",
    "\n",
    "        def handle_windows_exit(signum: int, frame: Optional[FrameType]) -> None:\n",
    "            self.should_exit = True\n",
    "\n",
    "        def handle_exit(sig: int) -> None:\n",
    "            self.should_exit = True\n",
    "\n",
    "        for sig in HANDLED_SIGNALS:\n",
    "            if platform.system() == \"Windows\":\n",
    "                signal.signal(sig, handle_windows_exit)\n",
    "            else:\n",
    "                loop.add_signal_handler(sig, handle_exit, sig)\n",
    "\n",
    "    async def _main_loop(self) -> None:\n",
    "        \"\"\"\n",
    "        Main loop for the FastKafka application server process.\n",
    "        \"\"\"\n",
    "        while not self.should_exit:\n",
    "            await asyncio.sleep(0.1)"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": null,
   "id": "ced70e01",
   "metadata": {},
   "outputs": [],
   "source": [
    "# | export\n",
    "\n",
    "_app = typer.Typer()\n",
    "\n",
    "\n",
    "@_app.command()\n",
    "def run_fastkafka_server_process(\n",
    "    app: str = typer.Argument(\n",
    "        ...,\n",
    "        help=\"Input in the form of 'path:app', where **path** is the path to a python file and **app** is an object of type **FastKafka**.\",\n",
    "    ),\n",
    "    kafka_broker: str = typer.Option(\n",
    "        ...,\n",
    "        help=\"Kafka broker, one of the keys of the kafka_brokers dictionary passed in the constructor of FastKafka class.\",\n",
    "    ),\n",
    ") -> None:\n",
    "    ServerProcess(app, kafka_broker).run()"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": null,
   "id": "e1de33fe",
   "metadata": {},
   "outputs": [
    {
     "name": "stdout",
     "output_type": "stream",
     "text": [
      "WARNING: make sure you save the notebook before running this cell\n",
      "\n",
      "Exporting and installing the new version of the CLI command...\n",
      "Defaulting to user installation because normal site-packages is not writeable\n",
      "Obtaining file:///work/fastkafka\n",
      "  Preparing metadata (setup.py): started\n",
      "  Preparing metadata (setup.py): finished with status 'done'\n",
      "Requirement already satisfied: aiokafka>=0.8.0 in /home/kumaran/.local/lib/python3.11/site-packages (from fastkafka==0.8.0) (0.8.1)\n",
      "Requirement already satisfied: anyio>=3.0 in /usr/local/lib/python3.11/dist-packages (from fastkafka==0.8.0) (3.7.0)\n",
      "Requirement already satisfied: asyncer>=0.0.2 in /home/kumaran/.local/lib/python3.11/site-packages (from fastkafka==0.8.0) (0.0.2)\n",
      "Requirement already satisfied: docstring-parser>=0.15 in /home/kumaran/.local/lib/python3.11/site-packages (from fastkafka==0.8.0) (0.15)\n",
      "Requirement already satisfied: nest-asyncio>=1.5.6 in /usr/local/lib/python3.11/dist-packages (from fastkafka==0.8.0) (1.5.6)\n",
      "Requirement already satisfied: pydantic>=2.0 in /home/kumaran/.local/lib/python3.11/site-packages (from fastkafka==0.8.0) (2.0.2)\n",
      "Requirement already satisfied: tqdm>=4.62 in /home/kumaran/.local/lib/python3.11/site-packages (from fastkafka==0.8.0) (4.65.0)\n",
      "Requirement already satisfied: typer>=0.7.0 in /home/kumaran/.local/lib/python3.11/site-packages (from fastkafka==0.8.0) (0.9.0)\n",
      "Requirement already satisfied: async-timeout in /home/kumaran/.local/lib/python3.11/site-packages (from aiokafka>=0.8.0->fastkafka==0.8.0) (4.0.2)\n",
      "Requirement already satisfied: kafka-python>=2.0.2 in /home/kumaran/.local/lib/python3.11/site-packages (from aiokafka>=0.8.0->fastkafka==0.8.0) (2.0.2)\n",
      "Requirement already satisfied: packaging in /usr/local/lib/python3.11/dist-packages (from aiokafka>=0.8.0->fastkafka==0.8.0) (23.1)\n",
      "Requirement already satisfied: idna>=2.8 in /usr/local/lib/python3.11/dist-packages (from anyio>=3.0->fastkafka==0.8.0) (3.4)\n",
      "Requirement already satisfied: sniffio>=1.1 in /usr/local/lib/python3.11/dist-packages (from anyio>=3.0->fastkafka==0.8.0) (1.3.0)\n",
      "Requirement already satisfied: annotated-types>=0.4.0 in /home/kumaran/.local/lib/python3.11/site-packages (from pydantic>=2.0->fastkafka==0.8.0) (0.5.0)\n",
      "Requirement already satisfied: pydantic-core==2.1.2 in /home/kumaran/.local/lib/python3.11/site-packages (from pydantic>=2.0->fastkafka==0.8.0) (2.1.2)\n",
      "Requirement already satisfied: typing-extensions>=4.6.1 in /home/kumaran/.local/lib/python3.11/site-packages (from pydantic>=2.0->fastkafka==0.8.0) (4.7.1)\n",
      "Requirement already satisfied: click<9.0.0,>=7.1.1 in /home/kumaran/.local/lib/python3.11/site-packages (from typer>=0.7.0->fastkafka==0.8.0) (8.1.4)\n"
     ]
    },
    {
     "name": "stderr",
     "output_type": "stream",
     "text": [
      "WARNING: fastkafka 0.8.0 does not provide the extra 'all'\n"
     ]
    },
    {
     "name": "stdout",
     "output_type": "stream",
     "text": [
      "Installing collected packages: fastkafka\n",
      "  Attempting uninstall: fastkafka\n",
      "    Found existing installation: fastkafka 0.8.0\n",
      "    Uninstalling fastkafka-0.8.0:\n",
      "      Successfully uninstalled fastkafka-0.8.0\n",
      "  Running setup.py develop for fastkafka\n",
      "Successfully installed fastkafka-0.8.0\n",
      "ok\n"
     ]
    }
   ],
   "source": [
    "# | notest\n",
    "\n",
    "print(\"WARNING: make sure you save the notebook before running this cell\\n\")\n",
    "\n",
    "print(\"Exporting and installing the new version of the CLI command...\")\n",
    "await asyncio.create_subprocess_exec(\"nbdev_export\")\n",
    "export_process = await asyncio.create_subprocess_exec(\"nbdev_export\")\n",
    "await export_process.wait()\n",
    "assert export_process.returncode == 0\n",
    "\n",
    "install_process = await asyncio.create_subprocess_exec(\n",
    "    \"pip\", \"install\", \"-e\", \"..[all]\"\n",
    ")\n",
    "await install_process.wait()\n",
    "assert install_process.returncode == 0\n",
    "\n",
    "print(\"ok\")"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": null,
   "id": "a5b0fe1a",
   "metadata": {},
   "outputs": [
    {
     "data": {
      "text/html": [
       "<pre style=\"white-space:pre;overflow-x:auto;line-height:normal;font-family:Menlo,'DejaVu Sans Mono',consolas,'Courier New',monospace\"><span style=\"font-weight: bold\">                                                                                                                   </span>\n",
       "<span style=\"font-weight: bold\"> </span><span style=\"color: #808000; text-decoration-color: #808000; font-weight: bold\">Usage: </span><span style=\"font-weight: bold\">run-fastkafka-server-process [OPTIONS] APP                                                                 </span>\n",
       "<span style=\"font-weight: bold\">                                                                                                                   </span>\n",
       "</pre>\n"
      ],
      "text/plain": [
       "\u001b[1m                                                                                                                   \u001b[0m\n",
       "\u001b[1m \u001b[0m\u001b[1;33mUsage: \u001b[0m\u001b[1mrun-fastkafka-server-process [OPTIONS] APP\u001b[0m\u001b[1m                                                                \u001b[0m\u001b[1m \u001b[0m\n",
       "\u001b[1m                                                                                                                   \u001b[0m\n"
      ]
     },
     "metadata": {},
     "output_type": "display_data"
    },
    {
     "data": {
      "text/html": [
       "<pre style=\"white-space:pre;overflow-x:auto;line-height:normal;font-family:Menlo,'DejaVu Sans Mono',consolas,'Courier New',monospace\"><span style=\"color: #7f7f7f; text-decoration-color: #7f7f7f\">╭─ Arguments ─────────────────────────────────────────────────────────────────────────────────────────────────────╮</span>\n",
       "<span style=\"color: #7f7f7f; text-decoration-color: #7f7f7f\">│</span> <span style=\"color: #800000; text-decoration-color: #800000\">*</span>    app      <span style=\"color: #808000; text-decoration-color: #808000; font-weight: bold\">TEXT</span>  Input in the form of 'path:app', where **path** is the path to a python file and **app** is <span style=\"color: #7f7f7f; text-decoration-color: #7f7f7f\">│</span>\n",
       "<span style=\"color: #7f7f7f; text-decoration-color: #7f7f7f\">│</span>                     an object of type **FastKafka**.                                                            <span style=\"color: #7f7f7f; text-decoration-color: #7f7f7f\">│</span>\n",
       "<span style=\"color: #7f7f7f; text-decoration-color: #7f7f7f\">│</span>                     <span style=\"color: #7f7f7f; text-decoration-color: #7f7f7f\">[default: None]                                                                            </span> <span style=\"color: #7f7f7f; text-decoration-color: #7f7f7f\">│</span>\n",
       "<span style=\"color: #7f7f7f; text-decoration-color: #7f7f7f\">│</span>                     <span style=\"color: #bf7f7f; text-decoration-color: #bf7f7f\">[required]                                                                                 </span> <span style=\"color: #7f7f7f; text-decoration-color: #7f7f7f\">│</span>\n",
       "<span style=\"color: #7f7f7f; text-decoration-color: #7f7f7f\">╰─────────────────────────────────────────────────────────────────────────────────────────────────────────────────╯</span>\n",
       "</pre>\n"
      ],
      "text/plain": [
       "\u001b[2m╭─\u001b[0m\u001b[2m Arguments \u001b[0m\u001b[2m────────────────────────────────────────────────────────────────────────────────────────────────────\u001b[0m\u001b[2m─╮\u001b[0m\n",
       "\u001b[2m│\u001b[0m \u001b[31m*\u001b[0m    app      \u001b[1;33mTEXT\u001b[0m  Input in the form of 'path:app', where **path** is the path to a python file and **app** is \u001b[2m│\u001b[0m\n",
       "\u001b[2m│\u001b[0m                     an object of type **FastKafka**.                                                            \u001b[2m│\u001b[0m\n",
       "\u001b[2m│\u001b[0m                     \u001b[2m[default: None]                                                                            \u001b[0m \u001b[2m│\u001b[0m\n",
       "\u001b[2m│\u001b[0m                     \u001b[2;31m[required]                                                                                 \u001b[0m \u001b[2m│\u001b[0m\n",
       "\u001b[2m╰─────────────────────────────────────────────────────────────────────────────────────────────────────────────────╯\u001b[0m\n"
      ]
     },
     "metadata": {},
     "output_type": "display_data"
    },
    {
     "data": {
      "text/html": [
       "<pre style=\"white-space:pre;overflow-x:auto;line-height:normal;font-family:Menlo,'DejaVu Sans Mono',consolas,'Courier New',monospace\"><span style=\"color: #7f7f7f; text-decoration-color: #7f7f7f\">╭─ Options ───────────────────────────────────────────────────────────────────────────────────────────────────────╮</span>\n",
       "<span style=\"color: #7f7f7f; text-decoration-color: #7f7f7f\">│</span> <span style=\"color: #800000; text-decoration-color: #800000\">*</span>  <span style=\"color: #008080; text-decoration-color: #008080; font-weight: bold\">--kafka-broker</span>              <span style=\"color: #808000; text-decoration-color: #808000; font-weight: bold\">TEXT</span>  Kafka broker, one of the keys of the kafka_brokers dictionary passed in    <span style=\"color: #7f7f7f; text-decoration-color: #7f7f7f\">│</span>\n",
       "<span style=\"color: #7f7f7f; text-decoration-color: #7f7f7f\">│</span>                                      the constructor of FastKafka class.                                        <span style=\"color: #7f7f7f; text-decoration-color: #7f7f7f\">│</span>\n",
       "<span style=\"color: #7f7f7f; text-decoration-color: #7f7f7f\">│</span>                                      <span style=\"color: #7f7f7f; text-decoration-color: #7f7f7f\">[default: None]                                                           </span> <span style=\"color: #7f7f7f; text-decoration-color: #7f7f7f\">│</span>\n",
       "<span style=\"color: #7f7f7f; text-decoration-color: #7f7f7f\">│</span>                                      <span style=\"color: #bf7f7f; text-decoration-color: #bf7f7f\">[required]                                                                </span> <span style=\"color: #7f7f7f; text-decoration-color: #7f7f7f\">│</span>\n",
       "<span style=\"color: #7f7f7f; text-decoration-color: #7f7f7f\">│</span>    <span style=\"color: #008080; text-decoration-color: #008080; font-weight: bold\">--install-completion</span>        <span style=\"color: #808000; text-decoration-color: #808000; font-weight: bold\">    </span>  Install completion for the current shell.                                  <span style=\"color: #7f7f7f; text-decoration-color: #7f7f7f\">│</span>\n",
       "<span style=\"color: #7f7f7f; text-decoration-color: #7f7f7f\">│</span>    <span style=\"color: #008080; text-decoration-color: #008080; font-weight: bold\">--show-completion</span>           <span style=\"color: #808000; text-decoration-color: #808000; font-weight: bold\">    </span>  Show completion for the current shell, to copy it or customize the         <span style=\"color: #7f7f7f; text-decoration-color: #7f7f7f\">│</span>\n",
       "<span style=\"color: #7f7f7f; text-decoration-color: #7f7f7f\">│</span>                                      installation.                                                              <span style=\"color: #7f7f7f; text-decoration-color: #7f7f7f\">│</span>\n",
       "<span style=\"color: #7f7f7f; text-decoration-color: #7f7f7f\">│</span>    <span style=\"color: #008080; text-decoration-color: #008080; font-weight: bold\">--help</span>                      <span style=\"color: #808000; text-decoration-color: #808000; font-weight: bold\">    </span>  Show this message and exit.                                                <span style=\"color: #7f7f7f; text-decoration-color: #7f7f7f\">│</span>\n",
       "<span style=\"color: #7f7f7f; text-decoration-color: #7f7f7f\">╰─────────────────────────────────────────────────────────────────────────────────────────────────────────────────╯</span>\n",
       "</pre>\n"
      ],
      "text/plain": [
       "\u001b[2m╭─\u001b[0m\u001b[2m Options \u001b[0m\u001b[2m──────────────────────────────────────────────────────────────────────────────────────────────────────\u001b[0m\u001b[2m─╮\u001b[0m\n",
       "\u001b[2m│\u001b[0m \u001b[31m*\u001b[0m  \u001b[1;36m-\u001b[0m\u001b[1;36m-kafka\u001b[0m\u001b[1;36m-broker\u001b[0m              \u001b[1;33mTEXT\u001b[0m  Kafka broker, one of the keys of the kafka_brokers dictionary passed in    \u001b[2m│\u001b[0m\n",
       "\u001b[2m│\u001b[0m                                      the constructor of FastKafka class.                                        \u001b[2m│\u001b[0m\n",
       "\u001b[2m│\u001b[0m                                      \u001b[2m[default: None]                                                           \u001b[0m \u001b[2m│\u001b[0m\n",
       "\u001b[2m│\u001b[0m                                      \u001b[2;31m[required]                                                                \u001b[0m \u001b[2m│\u001b[0m\n",
       "\u001b[2m│\u001b[0m    \u001b[1;36m-\u001b[0m\u001b[1;36m-install\u001b[0m\u001b[1;36m-completion\u001b[0m        \u001b[1;33m    \u001b[0m  Install completion for the current shell.                                  \u001b[2m│\u001b[0m\n",
       "\u001b[2m│\u001b[0m    \u001b[1;36m-\u001b[0m\u001b[1;36m-show\u001b[0m\u001b[1;36m-completion\u001b[0m           \u001b[1;33m    \u001b[0m  Show completion for the current shell, to copy it or customize the         \u001b[2m│\u001b[0m\n",
       "\u001b[2m│\u001b[0m                                      installation.                                                              \u001b[2m│\u001b[0m\n",
       "\u001b[2m│\u001b[0m    \u001b[1;36m-\u001b[0m\u001b[1;36m-help\u001b[0m                      \u001b[1;33m    \u001b[0m  Show this message and exit.                                                \u001b[2m│\u001b[0m\n",
       "\u001b[2m╰─────────────────────────────────────────────────────────────────────────────────────────────────────────────────╯\u001b[0m\n"
      ]
     },
     "metadata": {},
     "output_type": "display_data"
    }
   ],
   "source": [
    "runner = CliRunner()\n",
    "result = runner.invoke(_app, [\"run_fastkafka_server_process\", \"--help\"])"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": null,
   "id": "52e1c583",
   "metadata": {},
   "outputs": [],
   "source": [
    "# | export\n",
    "\n",
    "\n",
    "async def run_fastkafka_server(num_workers: int, app: str, kafka_broker: str) -> None:\n",
    "    \"\"\"\n",
    "    Runs the FastKafka server with multiple worker processes.\n",
    "\n",
    "    Args:\n",
    "        num_workers (int): Number of FastKafka instances to run.\n",
    "        app (str): Input in the form of 'path:app', where **path** is the path to a python file and **app** is an object of type **FastKafka**.\n",
    "        kafka_broker (str): Kafka broker, one of the keys of the kafka_brokers dictionary passed in the constructor of FastKafka class.\n",
    "    \"\"\"\n",
    "    loop = asyncio.get_event_loop()\n",
    "\n",
    "    HANDLED_SIGNALS = (\n",
    "        signal.SIGINT,  # Unix signal 2. Sent by Ctrl+C.\n",
    "        signal.SIGTERM,  # Unix signal 15. Sent by `kill <pid>`.\n",
    "    )\n",
    "    if platform.system() == \"Windows\":\n",
    "            HANDLED_SIGNALS = (*HANDLED_SIGNALS, signal.SIGBREAK) # type: ignore\n",
    "\n",
    "    d = {\"should_exit\": False}\n",
    "\n",
    "    def handle_windows_exit(\n",
    "        signum: int, frame: Optional[FrameType], d: Dict[str, bool] = d\n",
    "    ) -> None:\n",
    "        d[\"should_exit\"] = True\n",
    "\n",
    "    def handle_exit(sig: int, d: Dict[str, bool] = d) -> None:\n",
    "        d[\"should_exit\"] = True\n",
    "\n",
    "    for sig in HANDLED_SIGNALS:\n",
    "        if platform.system() == \"Windows\":\n",
    "            signal.signal(sig, handle_windows_exit)\n",
    "        else:\n",
    "            loop.add_signal_handler(sig, handle_exit, sig)\n",
    "\n",
    "    async with asyncer.create_task_group() as tg:\n",
    "        args = [\n",
    "            \"run_fastkafka_server_process\",\n",
    "            \"--kafka-broker\",\n",
    "            kafka_broker,\n",
    "            app,\n",
    "        ]\n",
    "        tasks = [\n",
    "            tg.soonify(asyncio.create_subprocess_exec)(\n",
    "                *args,\n",
    "                limit=1024*1024, # Set StreamReader buffer limit to 1MB\n",
    "                stdout=asyncio.subprocess.PIPE,\n",
    "                stdin=asyncio.subprocess.PIPE,\n",
    "            )\n",
    "            for i in range(num_workers)\n",
    "        ]\n",
    "\n",
    "    procs = [task.value for task in tasks]\n",
    "\n",
    "    async def log_output(\n",
    "        output: Optional[asyncio.StreamReader], pid: int, d: Dict[str, bool] = d\n",
    "    ) -> None:\n",
    "        if output is None:\n",
    "            raise RuntimeError(\"Expected StreamReader, got None. Is stdout piped?\")\n",
    "        while not output.at_eof():\n",
    "            try:\n",
    "                outs = await output.readline()\n",
    "            except ValueError:\n",
    "                typer.echo(f\"[{pid:03d}]: Failed to read log output\", nl=False)\n",
    "                continue\n",
    "            if outs != b\"\":\n",
    "                typer.echo(f\"[{pid:03d}]: \" + outs.decode(\"utf-8\").strip(), nl=False)\n",
    "\n",
    "    async with asyncer.create_task_group() as tg:\n",
    "        for proc in procs:\n",
    "            tg.soonify(log_output)(proc.stdout, proc.pid)\n",
    "\n",
    "        while not d[\"should_exit\"]:\n",
    "            await asyncio.sleep(0.2)\n",
    "\n",
    "        typer.echo(\"Starting process cleanup, this may take a few seconds...\")\n",
    "        for proc in procs:\n",
    "            tg.soonify(terminate_asyncio_process)(proc)\n",
    "\n",
    "    for proc in procs:\n",
    "        output, _ = await proc.communicate()\n",
    "        if output:\n",
    "            typer.echo(f\"[{proc.pid:03d}]: \" + output.decode(\"utf-8\").strip(), nl=False)\n",
    "\n",
    "    returncodes = [proc.returncode for proc in procs]\n",
    "    if not returncodes == [0] * len(procs):\n",
    "        typer.secho(\n",
    "            f\"Return codes are not all zero: {returncodes}\",\n",
    "            err=True,\n",
    "            fg=typer.colors.RED,\n",
    "        )\n",
    "        raise typer.Exit(1)"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": null,
   "id": "804d0df0",
   "metadata": {},
   "outputs": [],
   "source": [
    "# | export\n",
    "\n",
    "\n",
    "@contextmanager\n",
    "def run_in_process(\n",
    "    target: Callable[..., Any]\n",
    ") -> Generator[multiprocessing.Process, None, None]:\n",
    "    \"\"\"\n",
    "    Runs the target function in a separate process.\n",
    "\n",
    "    Args:\n",
    "        target (Callable[..., Any]): The function to run in a separate process.\n",
    "\n",
    "    Yields:\n",
    "        Generator[multiprocessing.Process, None, None]: A generator that yields the process object.\n",
    "    \"\"\"\n",
    "    p = multiprocessing.Process(target=target)\n",
    "    try:\n",
    "        p.start()\n",
    "        yield p\n",
    "    except Exception as e:\n",
    "        print(f\"Exception raised {e=}\")\n",
    "    finally:\n",
    "        p.terminate()\n",
    "        p.join()"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": null,
   "id": "f7c8cea8",
   "metadata": {},
   "outputs": [
    {
     "name": "stdout",
     "output_type": "stream",
     "text": [
      "[INFO] fastkafka._components.test_dependencies: Java is already installed.\n",
      "[INFO] fastkafka._components.test_dependencies: But not exported to PATH, exporting...\n",
      "[INFO] fastkafka._components.test_dependencies: Kafka is installed.\n",
      "[INFO] fastkafka._components.test_dependencies: But not exported to PATH, exporting...\n",
      "[INFO] fastkafka._testing.apache_kafka_broker: Starting zookeeper...\n",
      "[INFO] fastkafka._testing.apache_kafka_broker: Starting kafka...\n",
      "[INFO] fastkafka._testing.apache_kafka_broker: Local Kafka broker up and running on 127.0.0.1:10000\n",
      "[88210]: 23-07-08 16:11:08.142 [INFO] fastkafka._application.app: _create_producer() : created producer using the config: '{'bootstrap_servers': '127.0.0.1:10000'}'\n",
      "[88212]: 23-07-08 16:11:08.142 [INFO] fastkafka._application.app: _create_producer() : created producer using the config: '{'bootstrap_servers': '127.0.0.1:10000'}'\n",
      "[88214]: 23-07-08 16:11:08.142 [INFO] fastkafka._application.app: _create_producer() : created producer using the config: '{'bootstrap_servers': '127.0.0.1:10000'}'\n",
      "[88208]: 23-07-08 16:11:08.144 [INFO] fastkafka._application.app: _create_producer() : created producer using the config: '{'bootstrap_servers': '127.0.0.1:10000'}'\n",
      "[88214]: 23-07-08 16:11:08.169 [INFO] fastkafka._application.app: _create_producer() : created producer using the config: '{'bootstrap_servers': '127.0.0.1:10000'}'\n",
      "[88212]: 23-07-08 16:11:08.169 [INFO] fastkafka._application.app: _create_producer() : created producer using the config: '{'bootstrap_servers': '127.0.0.1:10000'}'\n",
      "[88210]: 23-07-08 16:11:08.169 [INFO] fastkafka._application.app: _create_producer() : created producer using the config: '{'bootstrap_servers': '127.0.0.1:10000'}'\n",
      "[88208]: 23-07-08 16:11:08.169 [INFO] fastkafka._application.app: _create_producer() : created producer using the config: '{'bootstrap_servers': '127.0.0.1:10000'}'\n",
      "[88208]: 23-07-08 16:11:08.176 [INFO] fastkafka._application.app: _create_producer() : created producer using the config: '{'bootstrap_servers': '127.0.0.1:10000'}'\n",
      "[88212]: 23-07-08 16:11:08.176 [INFO] fastkafka._application.app: _create_producer() : created producer using the config: '{'bootstrap_servers': '127.0.0.1:10000'}'\n",
      "[88210]: 23-07-08 16:11:08.176 [INFO] fastkafka._application.app: _create_producer() : created producer using the config: '{'bootstrap_servers': '127.0.0.1:10000'}'\n",
      "[88214]: 23-07-08 16:11:08.176 [INFO] fastkafka._application.app: _create_producer() : created producer using the config: '{'bootstrap_servers': '127.0.0.1:10000'}'\n",
      "[88208]: 23-07-08 16:11:08.182 [INFO] fastkafka._application.app: _create_producer() : created producer using the config: '{'bootstrap_servers': '127.0.0.1:10000'}'\n",
      "[88212]: 23-07-08 16:11:08.182 [INFO] fastkafka._application.app: _create_producer() : created producer using the config: '{'bootstrap_servers': '127.0.0.1:10000'}'\n",
      "[88210]: 23-07-08 16:11:08.182 [INFO] fastkafka._application.app: _create_producer() : created producer using the config: '{'bootstrap_servers': '127.0.0.1:10000'}'\n",
      "[88214]: 23-07-08 16:11:08.182 [INFO] fastkafka._application.app: _create_producer() : created producer using the config: '{'bootstrap_servers': '127.0.0.1:10000'}'\n",
      "[88212]: 23-07-08 16:11:08.189 [INFO] fastkafka._components.aiokafka_consumer_loop: aiokafka_consumer_loop() starting...\n",
      "[88210]: 23-07-08 16:11:08.189 [INFO] fastkafka._components.aiokafka_consumer_loop: aiokafka_consumer_loop() starting...\n",
      "[88214]: 23-07-08 16:11:08.189 [INFO] fastkafka._components.aiokafka_consumer_loop: aiokafka_consumer_loop() starting...\n",
      "[88208]: 23-07-08 16:11:08.189 [INFO] fastkafka._components.aiokafka_consumer_loop: aiokafka_consumer_loop() starting...\n",
      "[88212]: 23-07-08 16:11:08.189 [INFO] fastkafka._components.aiokafka_consumer_loop: aiokafka_consumer_loop(): Consumer created using the following parameters: {'group_id': '127.0.0.1:10000_group', 'auto_offset_reset': 'earliest', 'max_poll_records': 100, 'bootstrap_servers': '127.0.0.1:10000'}\n",
      "[88212]: 23-07-08 16:11:08.190 [INFO] fastkafka._components.aiokafka_consumer_loop: aiokafka_consumer_loop() starting...\n",
      "[88212]: 23-07-08 16:11:08.190 [INFO] fastkafka._components.aiokafka_consumer_loop: aiokafka_consumer_loop(): Consumer created using the following parameters: {'group_id': '127.0.0.1:10000_group', 'auto_offset_reset': 'earliest', 'max_poll_records': 100, 'bootstrap_servers': '127.0.0.1:10000'}\n",
      "[88214]: 23-07-08 16:11:08.189 [INFO] fastkafka._components.aiokafka_consumer_loop: aiokafka_consumer_loop(): Consumer created using the following parameters: {'group_id': '127.0.0.1:10000_group', 'auto_offset_reset': 'earliest', 'max_poll_records': 100, 'bootstrap_servers': '127.0.0.1:10000'}\n",
      "[88214]: 23-07-08 16:11:08.190 [INFO] fastkafka._components.aiokafka_consumer_loop: aiokafka_consumer_loop() starting...\n",
      "[88214]: 23-07-08 16:11:08.190 [INFO] fastkafka._components.aiokafka_consumer_loop: aiokafka_consumer_loop(): Consumer created using the following parameters: {'group_id': '127.0.0.1:10000_group', 'auto_offset_reset': 'earliest', 'max_poll_records': 100, 'bootstrap_servers': '127.0.0.1:10000'}\n",
      "[88210]: 23-07-08 16:11:08.189 [INFO] fastkafka._components.aiokafka_consumer_loop: aiokafka_consumer_loop(): Consumer created using the following parameters: {'group_id': '127.0.0.1:10000_group', 'auto_offset_reset': 'earliest', 'max_poll_records': 100, 'bootstrap_servers': '127.0.0.1:10000'}\n",
      "[88210]: 23-07-08 16:11:08.190 [INFO] fastkafka._components.aiokafka_consumer_loop: aiokafka_consumer_loop() starting...\n",
      "[88210]: 23-07-08 16:11:08.190 [INFO] fastkafka._components.aiokafka_consumer_loop: aiokafka_consumer_loop(): Consumer created using the following parameters: {'group_id': '127.0.0.1:10000_group', 'auto_offset_reset': 'earliest', 'max_poll_records': 100, 'bootstrap_servers': '127.0.0.1:10000'}\n",
      "[88208]: 23-07-08 16:11:08.189 [INFO] fastkafka._components.aiokafka_consumer_loop: aiokafka_consumer_loop(): Consumer created using the following parameters: {'group_id': '127.0.0.1:10000_group', 'auto_offset_reset': 'earliest', 'max_poll_records': 100, 'bootstrap_servers': '127.0.0.1:10000'}\n",
      "[88208]: 23-07-08 16:11:08.190 [INFO] fastkafka._components.aiokafka_consumer_loop: aiokafka_consumer_loop() starting...\n",
      "[88208]: 23-07-08 16:11:08.190 [INFO] fastkafka._components.aiokafka_consumer_loop: aiokafka_consumer_loop(): Consumer created using the following parameters: {'group_id': '127.0.0.1:10000_group', 'auto_offset_reset': 'earliest', 'max_poll_records': 100, 'bootstrap_servers': '127.0.0.1:10000'}\n",
      "[88212]: 23-07-08 16:11:08.198 [INFO] fastkafka._components.aiokafka_consumer_loop: aiokafka_consumer_loop(): Consumer started.\n",
      "[88212]: 23-07-08 16:11:08.199 [INFO] aiokafka.consumer.subscription_state: Updating subscribed topics to: frozenset({'realitime_data'})\n",
      "[88212]: 23-07-08 16:11:08.199 [INFO] aiokafka.consumer.consumer: Subscribed to topic(s): {'realitime_data'}\n",
      "[88212]: 23-07-08 16:11:08.199 [INFO] fastkafka._components.aiokafka_consumer_loop: aiokafka_consumer_loop(): Consumer subscribed.\n",
      "[88212]: 23-07-08 16:11:08.204 [INFO] fastkafka._components.aiokafka_consumer_loop: aiokafka_consumer_loop(): Consumer started.\n",
      "[88212]: 23-07-08 16:11:08.205 [INFO] aiokafka.consumer.subscription_state: Updating subscribed topics to: frozenset({'training_data'})\n",
      "[88212]: 23-07-08 16:11:08.205 [INFO] aiokafka.consumer.consumer: Subscribed to topic(s): {'training_data'}\n",
      "[88212]: 23-07-08 16:11:08.205 [INFO] fastkafka._components.aiokafka_consumer_loop: aiokafka_consumer_loop(): Consumer subscribed.\n",
      "[88210]: 23-07-08 16:11:08.199 [INFO] fastkafka._components.aiokafka_consumer_loop: aiokafka_consumer_loop(): Consumer started.\n",
      "[88210]: 23-07-08 16:11:08.199 [INFO] aiokafka.consumer.subscription_state: Updating subscribed topics to: frozenset({'training_data'})\n",
      "[88210]: 23-07-08 16:11:08.199 [INFO] aiokafka.consumer.consumer: Subscribed to topic(s): {'training_data'}\n",
      "[88210]: 23-07-08 16:11:08.199 [INFO] fastkafka._components.aiokafka_consumer_loop: aiokafka_consumer_loop(): Consumer subscribed.\n",
      "[88210]: 23-07-08 16:11:08.205 [INFO] fastkafka._components.aiokafka_consumer_loop: aiokafka_consumer_loop(): Consumer started.\n",
      "[88210]: 23-07-08 16:11:08.205 [INFO] aiokafka.consumer.subscription_state: Updating subscribed topics to: frozenset({'realitime_data'})\n"
     ]
    },
    {
     "name": "stdout",
     "output_type": "stream",
     "text": [
      "[88210]: 23-07-08 16:11:08.205 [INFO] aiokafka.consumer.consumer: Subscribed to topic(s): {'realitime_data'}\n",
      "[88210]: 23-07-08 16:11:08.205 [INFO] fastkafka._components.aiokafka_consumer_loop: aiokafka_consumer_loop(): Consumer subscribed.\n",
      "[88214]: 23-07-08 16:11:08.199 [INFO] fastkafka._components.aiokafka_consumer_loop: aiokafka_consumer_loop(): Consumer started.\n",
      "[88214]: 23-07-08 16:11:08.199 [INFO] aiokafka.consumer.subscription_state: Updating subscribed topics to: frozenset({'realitime_data'})\n",
      "[88214]: 23-07-08 16:11:08.199 [INFO] aiokafka.consumer.consumer: Subscribed to topic(s): {'realitime_data'}\n",
      "[88214]: 23-07-08 16:11:08.199 [INFO] fastkafka._components.aiokafka_consumer_loop: aiokafka_consumer_loop(): Consumer subscribed.\n",
      "[88214]: 23-07-08 16:11:08.204 [INFO] fastkafka._components.aiokafka_consumer_loop: aiokafka_consumer_loop(): Consumer started.\n",
      "[88214]: 23-07-08 16:11:08.205 [INFO] aiokafka.consumer.subscription_state: Updating subscribed topics to: frozenset({'training_data'})\n",
      "[88214]: 23-07-08 16:11:08.205 [INFO] aiokafka.consumer.consumer: Subscribed to topic(s): {'training_data'}\n",
      "[88214]: 23-07-08 16:11:08.205 [INFO] fastkafka._components.aiokafka_consumer_loop: aiokafka_consumer_loop(): Consumer subscribed.\n",
      "[88208]: 23-07-08 16:11:08.200 [INFO] fastkafka._components.aiokafka_consumer_loop: aiokafka_consumer_loop(): Consumer started.\n",
      "[88208]: 23-07-08 16:11:08.200 [INFO] aiokafka.consumer.subscription_state: Updating subscribed topics to: frozenset({'realitime_data'})\n",
      "[88208]: 23-07-08 16:11:08.200 [INFO] aiokafka.consumer.consumer: Subscribed to topic(s): {'realitime_data'}\n",
      "[88208]: 23-07-08 16:11:08.200 [INFO] fastkafka._components.aiokafka_consumer_loop: aiokafka_consumer_loop(): Consumer subscribed.\n",
      "[88208]: 23-07-08 16:11:08.205 [INFO] fastkafka._components.aiokafka_consumer_loop: aiokafka_consumer_loop(): Consumer started.\n",
      "[88208]: 23-07-08 16:11:08.206 [INFO] aiokafka.consumer.subscription_state: Updating subscribed topics to: frozenset({'training_data'})\n",
      "[88208]: 23-07-08 16:11:08.207 [INFO] aiokafka.consumer.consumer: Subscribed to topic(s): {'training_data'}\n",
      "[88208]: 23-07-08 16:11:08.207 [INFO] fastkafka._components.aiokafka_consumer_loop: aiokafka_consumer_loop(): Consumer subscribed.\n",
      "[88212]: 23-07-08 16:11:08.220 [ERROR] aiokafka.consumer.group_coordinator: Group Coordinator Request failed: [Error 15] CoordinatorNotAvailableError\n",
      "[88212]: 23-07-08 16:11:08.221 [ERROR] aiokafka.consumer.group_coordinator: Group Coordinator Request failed: [Error 15] CoordinatorNotAvailableError\n",
      "[88212]: 23-07-08 16:11:08.227 [ERROR] aiokafka.cluster: Topic realitime_data not found in cluster metadata\n",
      "[88212]: 23-07-08 16:11:08.228 [ERROR] aiokafka.cluster: Topic training_data not found in cluster metadata\n",
      "[88214]: 23-07-08 16:11:08.220 [ERROR] aiokafka.consumer.group_coordinator: Group Coordinator Request failed: [Error 15] CoordinatorNotAvailableError\n",
      "[88214]: 23-07-08 16:11:08.221 [ERROR] aiokafka.consumer.group_coordinator: Group Coordinator Request failed: [Error 15] CoordinatorNotAvailableError\n",
      "[88214]: 23-07-08 16:11:08.228 [ERROR] aiokafka.cluster: Topic realitime_data not found in cluster metadata\n",
      "[88210]: 23-07-08 16:11:08.220 [ERROR] aiokafka.consumer.group_coordinator: Group Coordinator Request failed: [Error 15] CoordinatorNotAvailableError\n",
      "[88210]: 23-07-08 16:11:08.221 [ERROR] aiokafka.consumer.group_coordinator: Group Coordinator Request failed: [Error 15] CoordinatorNotAvailableError\n",
      "[88210]: 23-07-08 16:11:08.228 [ERROR] aiokafka.cluster: Topic training_data not found in cluster metadata\n",
      "[88208]: 23-07-08 16:11:08.220 [ERROR] aiokafka.consumer.group_coordinator: Group Coordinator Request failed: [Error 15] CoordinatorNotAvailableError\n",
      "[88208]: 23-07-08 16:11:08.227 [ERROR] aiokafka.cluster: Topic realitime_data not found in cluster metadata\n",
      "[88208]: 23-07-08 16:11:08.264 [ERROR] aiokafka.consumer.group_coordinator: Group Coordinator Request failed: [Error 15] CoordinatorNotAvailableError\n",
      "[88214]: 23-07-08 16:11:08.264 [WARNING] aiokafka.cluster: Topic training_data is not available during auto-create initialization\n",
      "[88210]: 23-07-08 16:11:08.264 [WARNING] aiokafka.cluster: Topic realitime_data is not available during auto-create initialization\n",
      "[88208]: 23-07-08 16:11:08.268 [WARNING] aiokafka.cluster: Topic training_data is not available during auto-create initialization\n",
      "[88214]: 23-07-08 16:11:08.330 [ERROR] aiokafka.consumer.group_coordinator: Group Coordinator Request failed: [Error 15] CoordinatorNotAvailableError\n",
      "[88210]: 23-07-08 16:11:08.330 [ERROR] aiokafka.consumer.group_coordinator: Group Coordinator Request failed: [Error 15] CoordinatorNotAvailableError\n",
      "[88212]: 23-07-08 16:11:08.330 [ERROR] aiokafka.consumer.group_coordinator: Group Coordinator Request failed: [Error 15] CoordinatorNotAvailableError\n",
      "[88212]: 23-07-08 16:11:08.330 [ERROR] aiokafka.consumer.group_coordinator: Group Coordinator Request failed: [Error 15] CoordinatorNotAvailableError\n",
      "[88212]: 23-07-08 16:11:08.331 [ERROR] aiokafka.cluster: Topic realitime_data not found in cluster metadata\n",
      "[88212]: 23-07-08 16:11:08.331 [ERROR] aiokafka.cluster: Topic training_data not found in cluster metadata\n",
      "[88208]: 23-07-08 16:11:08.339 [ERROR] aiokafka.consumer.group_coordinator: Group Coordinator Request failed: [Error 15] CoordinatorNotAvailableError\n",
      "[88210]: 23-07-08 16:11:08.339 [WARNING] aiokafka.cluster: Topic training_data is not available during auto-create initialization\n",
      "[88214]: 23-07-08 16:11:08.339 [WARNING] aiokafka.cluster: Topic realitime_data is not available during auto-create initialization\n",
      "[88208]: 23-07-08 16:11:08.344 [WARNING] aiokafka.cluster: Topic realitime_data is not available during auto-create initialization\n",
      "[88210]: 23-07-08 16:11:08.367 [ERROR] aiokafka.consumer.group_coordinator: Group Coordinator Request failed: [Error 15] CoordinatorNotAvailableError\n",
      "[88214]: 23-07-08 16:11:08.368 [ERROR] aiokafka.consumer.group_coordinator: Group Coordinator Request failed: [Error 15] CoordinatorNotAvailableError\n",
      "[88210]: 23-07-08 16:11:08.370 [WARNING] aiokafka.cluster: Topic realitime_data is not available during auto-create initialization\n",
      "[88214]: 23-07-08 16:11:08.370 [WARNING] aiokafka.cluster: Topic training_data is not available during auto-create initialization\n",
      "[88208]: 23-07-08 16:11:08.374 [ERROR] aiokafka.consumer.group_coordinator: Group Coordinator Request failed: [Error 15] CoordinatorNotAvailableError\n",
      "[88208]: 23-07-08 16:11:08.377 [WARNING] aiokafka.cluster: Topic training_data is not available during auto-create initialization\n",
      "[88212]: 23-07-08 16:11:08.434 [ERROR] aiokafka.consumer.group_coordinator: Group Coordinator Request failed: [Error 15] CoordinatorNotAvailableError\n",
      "[88212]: 23-07-08 16:11:08.436 [ERROR] aiokafka.consumer.group_coordinator: Group Coordinator Request failed: [Error 15] CoordinatorNotAvailableError\n",
      "[88212]: 23-07-08 16:11:08.436 [WARNING] aiokafka.cluster: Topic training_data is not available during auto-create initialization\n",
      "[88212]: 23-07-08 16:11:08.439 [WARNING] aiokafka.cluster: Topic realitime_data is not available during auto-create initialization\n",
      "[88210]: 23-07-08 16:11:08.441 [ERROR] aiokafka.consumer.group_coordinator: Group Coordinator Request failed: [Error 15] CoordinatorNotAvailableError\n",
      "[88214]: 23-07-08 16:11:08.443 [ERROR] aiokafka.consumer.group_coordinator: Group Coordinator Request failed: [Error 15] CoordinatorNotAvailableError\n",
      "[88210]: 23-07-08 16:11:08.444 [WARNING] aiokafka.cluster: Topic training_data is not available during auto-create initialization\n",
      "[88214]: 23-07-08 16:11:08.446 [WARNING] aiokafka.cluster: Topic realitime_data is not available during auto-create initialization\n",
      "[88208]: 23-07-08 16:11:08.448 [ERROR] aiokafka.consumer.group_coordinator: Group Coordinator Request failed: [Error 15] CoordinatorNotAvailableError\n",
      "[88208]: 23-07-08 16:11:08.450 [WARNING] aiokafka.cluster: Topic realitime_data is not available during auto-create initialization\n",
      "[88210]: 23-07-08 16:11:08.472 [ERROR] aiokafka.consumer.group_coordinator: Group Coordinator Request failed: [Error 15] CoordinatorNotAvailableError\n",
      "[88214]: 23-07-08 16:11:08.475 [ERROR] aiokafka.consumer.group_coordinator: Group Coordinator Request failed: [Error 15] CoordinatorNotAvailableError\n"
     ]
    },
    {
     "name": "stdout",
     "output_type": "stream",
     "text": [
      "[88210]: 23-07-08 16:11:08.476 [WARNING] aiokafka.cluster: Topic realitime_data is not available during auto-create initialization\n",
      "[88214]: 23-07-08 16:11:08.479 [WARNING] aiokafka.cluster: Topic training_data is not available during auto-create initialization\n",
      "[88208]: 23-07-08 16:11:08.481 [ERROR] aiokafka.consumer.group_coordinator: Group Coordinator Request failed: [Error 15] CoordinatorNotAvailableError\n",
      "[88208]: 23-07-08 16:11:08.484 [WARNING] aiokafka.cluster: Topic training_data is not available during auto-create initialization\n",
      "[88212]: 23-07-08 16:11:08.540 [ERROR] aiokafka.consumer.group_coordinator: Group Coordinator Request failed: [Error 15] CoordinatorNotAvailableError\n",
      "[88212]: 23-07-08 16:11:08.543 [ERROR] aiokafka.consumer.group_coordinator: Group Coordinator Request failed: [Error 15] CoordinatorNotAvailableError\n",
      "[88212]: 23-07-08 16:11:08.543 [WARNING] aiokafka.cluster: Topic training_data is not available during auto-create initialization\n",
      "[88212]: 23-07-08 16:11:08.546 [WARNING] aiokafka.cluster: Topic realitime_data is not available during auto-create initialization\n",
      "[88210]: 23-07-08 16:11:08.546 [ERROR] aiokafka.consumer.group_coordinator: Group Coordinator Request failed: [Error 15] CoordinatorNotAvailableError\n",
      "[88210]: 23-07-08 16:11:08.548 [WARNING] aiokafka.cluster: Topic training_data is not available during auto-create initialization\n",
      "[88214]: 23-07-08 16:11:08.549 [ERROR] aiokafka.consumer.group_coordinator: Group Coordinator Request failed: [Error 15] CoordinatorNotAvailableError\n",
      "[88214]: 23-07-08 16:11:08.552 [WARNING] aiokafka.cluster: Topic realitime_data is not available during auto-create initialization\n",
      "[88208]: 23-07-08 16:11:08.556 [ERROR] aiokafka.consumer.group_coordinator: Group Coordinator Request failed: [Error 15] CoordinatorNotAvailableError\n",
      "[88208]: 23-07-08 16:11:08.558 [WARNING] aiokafka.cluster: Topic realitime_data is not available during auto-create initialization\n",
      "[88210]: 23-07-08 16:11:08.579 [ERROR] aiokafka.consumer.group_coordinator: Group Coordinator Request failed: [Error 15] CoordinatorNotAvailableError\n",
      "[88210]: 23-07-08 16:11:08.581 [WARNING] aiokafka.cluster: Topic realitime_data is not available during auto-create initialization\n",
      "[88214]: 23-07-08 16:11:08.582 [ERROR] aiokafka.consumer.group_coordinator: Group Coordinator Request failed: [Error 15] CoordinatorNotAvailableError\n",
      "[88214]: 23-07-08 16:11:08.584 [WARNING] aiokafka.cluster: Topic training_data is not available during auto-create initialization\n",
      "[88208]: 23-07-08 16:11:08.588 [ERROR] aiokafka.consumer.group_coordinator: Group Coordinator Request failed: [Error 15] CoordinatorNotAvailableError\n",
      "[88208]: 23-07-08 16:11:08.590 [WARNING] aiokafka.cluster: Topic training_data is not available during auto-create initialization\n",
      "[88212]: 23-07-08 16:11:08.646 [ERROR] aiokafka.consumer.group_coordinator: Group Coordinator Request failed: [Error 15] CoordinatorNotAvailableError\n",
      "[88212]: 23-07-08 16:11:08.648 [ERROR] aiokafka.consumer.group_coordinator: Group Coordinator Request failed: [Error 15] CoordinatorNotAvailableError\n",
      "[88212]: 23-07-08 16:11:08.649 [WARNING] aiokafka.cluster: Topic training_data is not available during auto-create initialization\n",
      "[88212]: 23-07-08 16:11:08.651 [WARNING] aiokafka.cluster: Topic realitime_data is not available during auto-create initialization\n",
      "[88210]: 23-07-08 16:11:08.651 [ERROR] aiokafka.consumer.group_coordinator: Group Coordinator Request failed: [Error 15] CoordinatorNotAvailableError\n",
      "[88210]: 23-07-08 16:11:08.654 [WARNING] aiokafka.cluster: Topic training_data is not available during auto-create initialization\n",
      "[88214]: 23-07-08 16:11:08.655 [ERROR] aiokafka.consumer.group_coordinator: Group Coordinator Request failed: [Error 15] CoordinatorNotAvailableError\n",
      "[88214]: 23-07-08 16:11:08.657 [WARNING] aiokafka.cluster: Topic realitime_data is not available during auto-create initialization\n",
      "[88208]: 23-07-08 16:11:08.661 [ERROR] aiokafka.consumer.group_coordinator: Group Coordinator Request failed: [Error 15] CoordinatorNotAvailableError\n",
      "[88208]: 23-07-08 16:11:08.663 [WARNING] aiokafka.cluster: Topic realitime_data is not available during auto-create initialization\n",
      "[88210]: 23-07-08 16:11:08.684 [ERROR] aiokafka.consumer.group_coordinator: Group Coordinator Request failed: [Error 15] CoordinatorNotAvailableError\n",
      "[88210]: 23-07-08 16:11:08.686 [WARNING] aiokafka.cluster: Topic realitime_data is not available during auto-create initialization\n",
      "[88214]: 23-07-08 16:11:08.686 [ERROR] aiokafka.consumer.group_coordinator: Group Coordinator Request failed: [Error 15] CoordinatorNotAvailableError\n",
      "[88214]: 23-07-08 16:11:08.689 [WARNING] aiokafka.cluster: Topic training_data is not available during auto-create initialization\n",
      "[88208]: 23-07-08 16:11:08.693 [ERROR] aiokafka.consumer.group_coordinator: Group Coordinator Request failed: [Error 15] CoordinatorNotAvailableError\n",
      "[88208]: 23-07-08 16:11:08.695 [WARNING] aiokafka.cluster: Topic training_data is not available during auto-create initialization\n",
      "[88212]: 23-07-08 16:11:08.753 [ERROR] aiokafka.consumer.group_coordinator: Group Coordinator Request failed: [Error 15] CoordinatorNotAvailableError\n",
      "[88212]: 23-07-08 16:11:08.756 [ERROR] aiokafka.consumer.group_coordinator: Group Coordinator Request failed: [Error 15] CoordinatorNotAvailableError\n",
      "[88212]: 23-07-08 16:11:08.757 [WARNING] aiokafka.cluster: Topic training_data is not available during auto-create initialization\n",
      "[88212]: 23-07-08 16:11:08.759 [WARNING] aiokafka.cluster: Topic realitime_data is not available during auto-create initialization\n",
      "[88210]: 23-07-08 16:11:08.760 [ERROR] aiokafka.consumer.group_coordinator: Group Coordinator Request failed: [Error 15] CoordinatorNotAvailableError\n",
      "[88214]: 23-07-08 16:11:08.763 [ERROR] aiokafka.consumer.group_coordinator: Group Coordinator Request failed: [Error 15] CoordinatorNotAvailableError\n",
      "[88210]: 23-07-08 16:11:08.764 [WARNING] aiokafka.cluster: Topic training_data is not available during auto-create initialization\n",
      "[88208]: 23-07-08 16:11:08.767 [ERROR] aiokafka.consumer.group_coordinator: Group Coordinator Request failed: [Error 15] CoordinatorNotAvailableError\n",
      "[88214]: 23-07-08 16:11:08.767 [WARNING] aiokafka.cluster: Topic realitime_data is not available during auto-create initialization\n",
      "[88208]: 23-07-08 16:11:08.770 [WARNING] aiokafka.cluster: Topic realitime_data is not available during auto-create initialization\n",
      "[88214]: 23-07-08 16:11:08.794 [INFO] aiokafka.consumer.group_coordinator: Discovered coordinator 0 for group 127.0.0.1:10000_group\n",
      "[88214]: 23-07-08 16:11:08.794 [INFO] aiokafka.consumer.group_coordinator: Revoking previously assigned partitions set() for group 127.0.0.1:10000_group\n",
      "[88210]: 23-07-08 16:11:08.794 [INFO] aiokafka.consumer.group_coordinator: Discovered coordinator 0 for group 127.0.0.1:10000_group\n",
      "[88210]: 23-07-08 16:11:08.795 [INFO] aiokafka.consumer.group_coordinator: Revoking previously assigned partitions set() for group 127.0.0.1:10000_group\n",
      "[88214]: 23-07-08 16:11:08.795 [INFO] aiokafka.consumer.group_coordinator: (Re-)joining group 127.0.0.1:10000_group\n",
      "[88210]: 23-07-08 16:11:08.795 [INFO] aiokafka.consumer.group_coordinator: (Re-)joining group 127.0.0.1:10000_group\n",
      "[88208]: 23-07-08 16:11:08.799 [INFO] aiokafka.consumer.group_coordinator: Discovered coordinator 0 for group 127.0.0.1:10000_group\n",
      "[88208]: 23-07-08 16:11:08.799 [INFO] aiokafka.consumer.group_coordinator: Revoking previously assigned partitions set() for group 127.0.0.1:10000_group\n",
      "[88208]: 23-07-08 16:11:08.800 [INFO] aiokafka.consumer.group_coordinator: (Re-)joining group 127.0.0.1:10000_group\n",
      "[88214]: 23-07-08 16:11:08.825 [INFO] aiokafka.consumer.group_coordinator: Joined group '127.0.0.1:10000_group' (generation 1) with member_id aiokafka-0.8.1-948daba9-9ad3-425f-ba79-8eeb737b877f\n",
      "[88208]: 23-07-08 16:11:08.826 [INFO] aiokafka.consumer.group_coordinator: Joined group '127.0.0.1:10000_group' (generation 1) with member_id aiokafka-0.8.1-ba7c4e0b-0df8-4a5a-87a6-eec33b0ff8de\n",
      "[88210]: 23-07-08 16:11:08.827 [INFO] aiokafka.consumer.group_coordinator: Joined group '127.0.0.1:10000_group' (generation 1) with member_id aiokafka-0.8.1-819b4212-60d8-488f-a9f6-3280c6ab2bd2\n"
     ]
    },
    {
     "name": "stdout",
     "output_type": "stream",
     "text": [
      "[88210]: 23-07-08 16:11:08.827 [INFO] aiokafka.consumer.group_coordinator: Elected group leader -- performing partition assignments using roundrobin\n",
      "[88210]: 23-07-08 16:11:08.829 [INFO] aiokafka.consumer.group_coordinator: Metadata for topic has changed from {} to {'training_data': 1, 'realitime_data': 1}. \n",
      "[88212]: 23-07-08 16:11:08.861 [INFO] aiokafka.consumer.group_coordinator: Discovered coordinator 0 for group 127.0.0.1:10000_group\n",
      "[88212]: 23-07-08 16:11:08.861 [INFO] aiokafka.consumer.group_coordinator: Revoking previously assigned partitions set() for group 127.0.0.1:10000_group\n",
      "[88212]: 23-07-08 16:11:08.861 [INFO] aiokafka.consumer.group_coordinator: (Re-)joining group 127.0.0.1:10000_group\n",
      "[88212]: 23-07-08 16:11:08.863 [INFO] aiokafka.consumer.group_coordinator: Discovered coordinator 0 for group 127.0.0.1:10000_group\n",
      "[88212]: 23-07-08 16:11:08.863 [INFO] aiokafka.consumer.group_coordinator: Revoking previously assigned partitions set() for group 127.0.0.1:10000_group\n",
      "[88212]: 23-07-08 16:11:08.863 [INFO] aiokafka.consumer.group_coordinator: (Re-)joining group 127.0.0.1:10000_group\n",
      "[88210]: 23-07-08 16:11:08.868 [INFO] aiokafka.consumer.group_coordinator: Discovered coordinator 0 for group 127.0.0.1:10000_group\n",
      "[88210]: 23-07-08 16:11:08.869 [INFO] aiokafka.consumer.group_coordinator: Revoking previously assigned partitions set() for group 127.0.0.1:10000_group\n",
      "[88210]: 23-07-08 16:11:08.869 [INFO] aiokafka.consumer.group_coordinator: (Re-)joining group 127.0.0.1:10000_group\n",
      "[88214]: 23-07-08 16:11:08.872 [INFO] aiokafka.consumer.group_coordinator: Discovered coordinator 0 for group 127.0.0.1:10000_group\n",
      "[88214]: 23-07-08 16:11:08.872 [INFO] aiokafka.consumer.group_coordinator: Revoking previously assigned partitions set() for group 127.0.0.1:10000_group\n",
      "[88214]: 23-07-08 16:11:08.872 [INFO] aiokafka.consumer.group_coordinator: (Re-)joining group 127.0.0.1:10000_group\n",
      "[88214]: 23-07-08 16:11:08.873 [INFO] aiokafka.consumer.group_coordinator: Successfully synced group 127.0.0.1:10000_group with generation 1\n",
      "[88214]: 23-07-08 16:11:08.873 [INFO] aiokafka.consumer.group_coordinator: Setting newly assigned partitions {TopicPartition(topic='training_data', partition=0)} for group 127.0.0.1:10000_group\n",
      "[88208]: 23-07-08 16:11:08.874 [INFO] aiokafka.consumer.group_coordinator: Successfully synced group 127.0.0.1:10000_group with generation 1\n",
      "[88208]: 23-07-08 16:11:08.874 [INFO] aiokafka.consumer.group_coordinator: Setting newly assigned partitions set() for group 127.0.0.1:10000_group\n",
      "[88208]: 23-07-08 16:11:08.875 [INFO] aiokafka.consumer.group_coordinator: Discovered coordinator 0 for group 127.0.0.1:10000_group\n",
      "[88208]: 23-07-08 16:11:08.875 [INFO] aiokafka.consumer.group_coordinator: Revoking previously assigned partitions set() for group 127.0.0.1:10000_group\n",
      "[88208]: 23-07-08 16:11:08.875 [INFO] aiokafka.consumer.group_coordinator: (Re-)joining group 127.0.0.1:10000_group\n",
      "[88210]: 23-07-08 16:11:08.875 [INFO] aiokafka.consumer.group_coordinator: Successfully synced group 127.0.0.1:10000_group with generation 1\n",
      "[88210]: 23-07-08 16:11:08.875 [INFO] aiokafka.consumer.group_coordinator: Setting newly assigned partitions {TopicPartition(topic='realitime_data', partition=0)} for group 127.0.0.1:10000_group\n",
      "[88208]: 23-07-08 16:11:11.877 [WARNING] aiokafka.consumer.group_coordinator: Heartbeat failed for group 127.0.0.1:10000_group because it is rebalancing\n",
      "[88214]: 23-07-08 16:11:11.877 [WARNING] aiokafka.consumer.group_coordinator: Heartbeat failed for group 127.0.0.1:10000_group because it is rebalancing\n",
      "[88210]: 23-07-08 16:11:11.877 [WARNING] aiokafka.consumer.group_coordinator: Heartbeat failed for group 127.0.0.1:10000_group because it is rebalancing\n",
      "[88208]: 23-07-08 16:11:11.877 [INFO] aiokafka.consumer.group_coordinator: Revoking previously assigned partitions frozenset() for group 127.0.0.1:10000_group\n",
      "[88208]: 23-07-08 16:11:11.877 [INFO] aiokafka.consumer.group_coordinator: (Re-)joining group 127.0.0.1:10000_group\n",
      "[88210]: 23-07-08 16:11:11.890 [INFO] aiokafka.consumer.group_coordinator: Revoking previously assigned partitions frozenset({TopicPartition(topic='realitime_data', partition=0)}) for group 127.0.0.1:10000_group\n",
      "[88214]: 23-07-08 16:11:11.890 [INFO] aiokafka.consumer.group_coordinator: Revoking previously assigned partitions frozenset({TopicPartition(topic='training_data', partition=0)}) for group 127.0.0.1:10000_group\n",
      "[88210]: 23-07-08 16:11:11.890 [INFO] aiokafka.consumer.group_coordinator: (Re-)joining group 127.0.0.1:10000_group\n",
      "[88214]: 23-07-08 16:11:11.890 [INFO] aiokafka.consumer.group_coordinator: (Re-)joining group 127.0.0.1:10000_group\n",
      "[88214]: 23-07-08 16:11:11.891 [INFO] aiokafka.consumer.group_coordinator: Joined group '127.0.0.1:10000_group' (generation 2) with member_id aiokafka-0.8.1-a02307ca-1976-4092-a63a-74eddd0aa625\n",
      "[88212]: 23-07-08 16:11:11.892 [INFO] aiokafka.consumer.group_coordinator: Joined group '127.0.0.1:10000_group' (generation 2) with member_id aiokafka-0.8.1-54597d06-fb19-4c8f-a916-95bd77587674\n",
      "[88212]: 23-07-08 16:11:11.892 [INFO] aiokafka.consumer.group_coordinator: Joined group '127.0.0.1:10000_group' (generation 2) with member_id aiokafka-0.8.1-9a2bca7b-0e90-403e-a418-d7df5022f1db\n",
      "[88210]: 23-07-08 16:11:11.892 [INFO] aiokafka.consumer.group_coordinator: Joined group '127.0.0.1:10000_group' (generation 2) with member_id aiokafka-0.8.1-7c7affa3-7e32-49d1-85b5-71066432532e\n",
      "[88210]: 23-07-08 16:11:11.893 [INFO] aiokafka.consumer.group_coordinator: Joined group '127.0.0.1:10000_group' (generation 2) with member_id aiokafka-0.8.1-819b4212-60d8-488f-a9f6-3280c6ab2bd2\n",
      "[88210]: 23-07-08 16:11:11.894 [INFO] aiokafka.consumer.group_coordinator: Elected group leader -- performing partition assignments using roundrobin\n",
      "[88208]: 23-07-08 16:11:11.893 [INFO] aiokafka.consumer.group_coordinator: Joined group '127.0.0.1:10000_group' (generation 2) with member_id aiokafka-0.8.1-ba7c4e0b-0df8-4a5a-87a6-eec33b0ff8de\n",
      "[88208]: 23-07-08 16:11:11.893 [INFO] aiokafka.consumer.group_coordinator: Joined group '127.0.0.1:10000_group' (generation 2) with member_id aiokafka-0.8.1-cf4acaab-4ebd-4010-a60c-147126dc8711\n",
      "[88214]: 23-07-08 16:11:11.894 [INFO] aiokafka.consumer.group_coordinator: Joined group '127.0.0.1:10000_group' (generation 2) with member_id aiokafka-0.8.1-948daba9-9ad3-425f-ba79-8eeb737b877f\n",
      "[88214]: 23-07-08 16:11:11.897 [INFO] aiokafka.consumer.group_coordinator: Successfully synced group 127.0.0.1:10000_group with generation 2\n",
      "[88214]: 23-07-08 16:11:11.897 [INFO] aiokafka.consumer.group_coordinator: Setting newly assigned partitions set() for group 127.0.0.1:10000_group\n",
      "[88214]: 23-07-08 16:11:11.898 [INFO] aiokafka.consumer.group_coordinator: Successfully synced group 127.0.0.1:10000_group with generation 2\n",
      "[88214]: 23-07-08 16:11:11.898 [INFO] aiokafka.consumer.group_coordinator: Setting newly assigned partitions set() for group 127.0.0.1:10000_group\n",
      "[88212]: 23-07-08 16:11:11.897 [INFO] aiokafka.consumer.group_coordinator: Successfully synced group 127.0.0.1:10000_group with generation 2\n",
      "[88212]: 23-07-08 16:11:11.897 [INFO] aiokafka.consumer.group_coordinator: Setting newly assigned partitions {TopicPartition(topic='realitime_data', partition=0)} for group 127.0.0.1:10000_group\n",
      "[88212]: 23-07-08 16:11:11.898 [INFO] aiokafka.consumer.group_coordinator: Successfully synced group 127.0.0.1:10000_group with generation 2\n",
      "[88212]: 23-07-08 16:11:11.898 [INFO] aiokafka.consumer.group_coordinator: Setting newly assigned partitions set() for group 127.0.0.1:10000_group\n",
      "[88210]: 23-07-08 16:11:11.898 [INFO] aiokafka.consumer.group_coordinator: Successfully synced group 127.0.0.1:10000_group with generation 2\n",
      "[88210]: 23-07-08 16:11:11.898 [INFO] aiokafka.consumer.group_coordinator: Setting newly assigned partitions {TopicPartition(topic='training_data', partition=0)} for group 127.0.0.1:10000_group\n",
      "[88210]: 23-07-08 16:11:11.899 [INFO] aiokafka.consumer.group_coordinator: Successfully synced group 127.0.0.1:10000_group with generation 2\n",
      "[88210]: 23-07-08 16:11:11.899 [INFO] aiokafka.consumer.group_coordinator: Setting newly assigned partitions set() for group 127.0.0.1:10000_group\n",
      "[88208]: 23-07-08 16:11:11.898 [INFO] aiokafka.consumer.group_coordinator: Successfully synced group 127.0.0.1:10000_group with generation 2\n"
     ]
    },
    {
     "name": "stdout",
     "output_type": "stream",
     "text": [
      "[88208]: 23-07-08 16:11:11.898 [INFO] aiokafka.consumer.group_coordinator: Setting newly assigned partitions set() for group 127.0.0.1:10000_group\n",
      "[88208]: 23-07-08 16:11:11.898 [INFO] aiokafka.consumer.group_coordinator: Successfully synced group 127.0.0.1:10000_group with generation 2\n",
      "[88208]: 23-07-08 16:11:11.899 [INFO] aiokafka.consumer.group_coordinator: Setting newly assigned partitions set() for group 127.0.0.1:10000_group\n",
      "Starting process cleanup, this may take a few seconds...\n",
      "[INFO] fastkafka._components._subprocess: terminate_asyncio_process(): Terminating the process 88208...\n",
      "[INFO] fastkafka._components._subprocess: terminate_asyncio_process(): Terminating the process 88210...\n",
      "[INFO] fastkafka._components._subprocess: terminate_asyncio_process(): Terminating the process 88212...\n",
      "[INFO] fastkafka._components._subprocess: terminate_asyncio_process(): Terminating the process 88214...\n",
      "[88214]: 23-07-08 16:11:23.704 [INFO] aiokafka.consumer.group_coordinator: LeaveGroup request succeeded\n",
      "[88214]: 23-07-08 16:11:23.704 [INFO] aiokafka.consumer.group_coordinator: LeaveGroup request succeeded\n",
      "[88214]: 23-07-08 16:11:23.705 [INFO] fastkafka._components.aiokafka_consumer_loop: aiokafka_consumer_loop(): Consumer stopped.\n",
      "[88214]: 23-07-08 16:11:23.705 [INFO] fastkafka._components.aiokafka_consumer_loop: aiokafka_consumer_loop() finished.\n",
      "[88214]: 23-07-08 16:11:23.705 [INFO] fastkafka._components.aiokafka_consumer_loop: aiokafka_consumer_loop(): Consumer stopped.\n",
      "[88214]: 23-07-08 16:11:23.705 [INFO] fastkafka._components.aiokafka_consumer_loop: aiokafka_consumer_loop() finished.\n",
      "[88208]: 23-07-08 16:11:23.706 [INFO] aiokafka.consumer.group_coordinator: LeaveGroup request succeeded\n",
      "[88208]: 23-07-08 16:11:23.706 [INFO] aiokafka.consumer.group_coordinator: LeaveGroup request succeeded\n",
      "[88208]: 23-07-08 16:11:23.706 [INFO] fastkafka._components.aiokafka_consumer_loop: aiokafka_consumer_loop(): Consumer stopped.\n",
      "[88208]: 23-07-08 16:11:23.706 [INFO] fastkafka._components.aiokafka_consumer_loop: aiokafka_consumer_loop() finished.\n",
      "[88208]: 23-07-08 16:11:23.707 [INFO] fastkafka._components.aiokafka_consumer_loop: aiokafka_consumer_loop(): Consumer stopped.\n",
      "[88208]: 23-07-08 16:11:23.707 [INFO] fastkafka._components.aiokafka_consumer_loop: aiokafka_consumer_loop() finished.\n",
      "[88212]: 23-07-08 16:11:23.712 [INFO] aiokafka.consumer.group_coordinator: LeaveGroup request succeeded\n",
      "[88212]: 23-07-08 16:11:23.712 [INFO] fastkafka._components.aiokafka_consumer_loop: aiokafka_consumer_loop(): Consumer stopped.\n",
      "[88212]: 23-07-08 16:11:23.712 [INFO] fastkafka._components.aiokafka_consumer_loop: aiokafka_consumer_loop() finished.\n",
      "[88210]: 23-07-08 16:11:23.714 [INFO] aiokafka.consumer.group_coordinator: LeaveGroup request succeeded\n",
      "[88210]: 23-07-08 16:11:23.714 [INFO] fastkafka._components.aiokafka_consumer_loop: aiokafka_consumer_loop(): Consumer stopped.\n",
      "[88210]: 23-07-08 16:11:23.714 [INFO] fastkafka._components.aiokafka_consumer_loop: aiokafka_consumer_loop() finished.\n",
      "[88212]: 23-07-08 16:11:23.715 [INFO] aiokafka.consumer.group_coordinator: LeaveGroup request succeeded\n",
      "[88212]: 23-07-08 16:11:23.716 [INFO] fastkafka._components.aiokafka_consumer_loop: aiokafka_consumer_loop(): Consumer stopped.\n",
      "[88212]: 23-07-08 16:11:23.716 [INFO] fastkafka._components.aiokafka_consumer_loop: aiokafka_consumer_loop() finished.\n",
      "[88210]: 23-07-08 16:11:23.719 [INFO] aiokafka.consumer.group_coordinator: LeaveGroup request succeeded\n",
      "[88210]: 23-07-08 16:11:23.719 [INFO] fastkafka._components.aiokafka_consumer_loop: aiokafka_consumer_loop(): Consumer stopped.\n",
      "[88210]: 23-07-08 16:11:23.719 [INFO] fastkafka._components.aiokafka_consumer_loop: aiokafka_consumer_loop() finished.\n",
      "[INFO] fastkafka._components._subprocess: terminate_asyncio_process(): Process 88214 terminated.\n",
      "[INFO] fastkafka._components._subprocess: terminate_asyncio_process(): Process 88208 terminated.\n",
      "[INFO] fastkafka._components._subprocess: terminate_asyncio_process(): Process 88212 terminated.\n",
      "[INFO] fastkafka._components._subprocess: terminate_asyncio_process(): Process 88210 terminated.\n",
      "[INFO] fastkafka._components._subprocess: terminate_asyncio_process(): Terminating the process 87739...\n",
      "[INFO] fastkafka._components._subprocess: terminate_asyncio_process(): Process 87739 terminated.\n",
      "[INFO] fastkafka._components._subprocess: terminate_asyncio_process(): Terminating the process 87358...\n",
      "[INFO] fastkafka._components._subprocess: terminate_asyncio_process(): Process 87358 terminated.\n",
      "ok\n"
     ]
    }
   ],
   "source": [
    "# | notest\n",
    "\n",
    "listener_port = 10000\n",
    "async with ApacheKafkaBroker(listener_port=listener_port) as bootstrap_server:\n",
    "    os.environ[\"KAFKA_HOSTNAME\"], os.environ[\"KAFKA_PORT\"] = bootstrap_server.split(\":\")\n",
    "\n",
    "    with generate_app_in_tmp() as app:\n",
    "\n",
    "        def run_fastkafka_server_test():\n",
    "            asyncio.run(run_fastkafka_server(4, app, \"localhost\"))\n",
    "\n",
    "        with run_in_process(run_fastkafka_server_test) as p:\n",
    "            sleep(15)\n",
    "\n",
    "        assert p.exitcode == 0, p.exitcode\n",
    "        p.close()\n",
    "\n",
    "print(\"ok\")"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": null,
   "id": "2d11d7eb",
   "metadata": {},
   "outputs": [],
   "source": []
  }
 ],
 "metadata": {
  "kernelspec": {
   "display_name": "python3",
   "language": "python",
   "name": "python3"
  }
 },
 "nbformat": 4,
 "nbformat_minor": 5
}
